package com.cloudlead.common.worker.web;

import com.cloudlead.common.lang.Messages;
import com.cloudlead.common.lang.ObjectStatus;
import com.cloudlead.common.lang.api.SimpleActionRequest;
import com.cloudlead.common.lang.api.SimpleActionResponse;
import com.cloudlead.common.lang.exception.RequestNotAcceptableException;
import com.cloudlead.common.lang.exception.ResourceNotFoundException;
import com.cloudlead.common.lang.serialize.JSON;
import com.cloudlead.common.lang.utils.ExceptionHelper;
import com.cloudlead.common.lang.utils.ObjectId;
import com.sun.enterprise.ee.cms.core.GroupManagementService;
import com.sun.enterprise.ee.cms.core.MessageSignal;
import com.sun.enterprise.ee.cms.impl.client.MessageActionFactoryImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by dean on 2015/8/4.
 */
public class RemoteServiceRpcClient implements RemoteServiceClient {
    private static final Logger LOGGER = LoggerFactory
            .getLogger(RemoteServiceRpcClient.class);

    public static final String RPC_REQUST_COMPONENT = "rpcRequstComponent";
    public static final String RPC_RESPONSE_COMPONENT = "rpcResponseComponent";

    private Map<String, RpcCallback> responseWaitQueue = new ConcurrentHashMap<>();
    private Map<String, Class> responseTypeQueue = new ConcurrentHashMap<>();

    private static RemoteServiceRpcClient singleInstance;

    private RemoteServiceRpcClient() {
        GroupManagementService gms = WorkerManager.getInstance().getGMS();
        gms.addActionFactory(new MessageActionFactoryImpl(notification -> {
            if (notification instanceof MessageSignal) {
                MessageSignal messageSignal = (MessageSignal) notification;
                String responseStr = new String(messageSignal.getMessage());
                LOGGER.debug("<== client received from [" + notification.getMemberToken() + "] response is :" + responseStr);
                SimpleActionResponse simpleActionResponse = JSON.toBean(responseStr, SimpleActionResponse.class);
                String requestId = simpleActionResponse.getRequestId();
                Class responseType = responseTypeQueue.get(requestId);
                simpleActionResponse = JSON.toBean(responseStr, SimpleActionResponse.class, responseType);
                RpcCallback rpcCallback = responseWaitQueue.get(requestId);
                if (simpleActionResponse.isSuccess()) {
                    rpcCallback.success(simpleActionResponse.getContent());
                } else {
                    //还原异常
                    rpcCallback.error(ExceptionHelper.toException(simpleActionResponse.getCode(), simpleActionResponse.getMessage()));
                }
                responseTypeQueue.remove(requestId);
                responseWaitQueue.remove(requestId);
            }
        }), RPC_RESPONSE_COMPONENT);
    }

    public static synchronized RemoteServiceRpcClient getInstance() {
        if (null == singleInstance) {
            singleInstance = new RemoteServiceRpcClient();
        }
        return singleInstance;
    }

    /**
     * 异步执行远程服务
     *
     * @param remoteServiceClass
     * @param method
     * @param params
     * @param rpcCallback
     */
    public void asyncExecuteAction(Class<?> remoteServiceClass, Method method, Object[] params, RpcCallback rpcCallback) throws Exception {
        String action = getAction(remoteServiceClass, method.getName());
        String worker = WorkerRegistry.getInstance().findWorker(action);
        if (null == worker) {
            throw new ResourceNotFoundException(Messages.getMessage("worker.NotFound"));
        }

        WorkerInfo workerInfo = WorkerRegistry.getInstance().getWorkerList().get(worker);
        if (null == workerInfo) {
            throw new ResourceNotFoundException(Messages.getMessage("worker.info.NotFound"));
        }

        if (ObjectStatus.disabled.equals(workerInfo.getStatus())) {
            throw new RequestNotAcceptableException(Messages.getMessage("worker.status.disable"));
        }

        SimpleActionRequest request = new SimpleActionRequest();
        String requestId = ObjectId.get();
        request.setRequestId(requestId);
        Parameter[] parameters = method.getParameters();
        List<String> paramNames = workerInfo.getActions().get(action);
        for (int i = 0; i < parameters.length; i++) {
            String name = paramNames.get(i);
            request.addContent(name, params[i]);
        }

        RemoteServiceRequest remoteServiceRequest = new RemoteServiceRequest(action, JSON.toString(request));
        GroupManagementService gms = WorkerManager.getInstance().getGMS();
        String requestStr = JSON.toString(remoteServiceRequest);
        LOGGER.debug("==> client send to [" + worker + "] request is :" + requestStr);
        gms.getGroupHandle().sendMessage(worker, RPC_REQUST_COMPONENT, requestStr.getBytes());
        responseTypeQueue.put(requestId, method.getReturnType());
        responseWaitQueue.put(requestId, rpcCallback);

    }

    /**
     * 同步执行远程服务
     *
     * @param remoteServiceClass
     * @param method
     * @param params
     * @return
     */
    public Object executeAction(Class<?> remoteServiceClass, Method method, Object[] params) throws Exception {
        final Object[] result = {null};
        RpcCallback rpcCallback = new RpcCallback() {
            @Override
            public void success(Object response) {
                result[0] = response;
            }

            @Override
            public void error(Exception error) {
                result[0] = error;
            }
        };
        asyncExecuteAction(remoteServiceClass, method, params, rpcCallback);
        while (null == result[0]) {
            try {
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if (result[0] instanceof Exception) {
            throw (Exception) result[0];
        }
        return result[0];
    }

    private static String getAction(Class<?> remoteServiceClass, String method) {
        return WorkerRegistry.getInstance().findModule(remoteServiceClass.getTypeName()) + "@" + method;
    }
}
